#include "./include/TransportLayer.h"

TransportLayer::TransportLayer(
    const std::unordered_map<std::string, std::string> &inputData)
    : readData(inputData) {

    hashTab.resize(Configuration::dataConf.begin()->HashGroupCount);
    db = std::make_shared<DataBase>();
}
TransportLayer::~TransportLayer() {
}

int TransportLayer::Init() {
    zlog_w("TransportLayer initialize");
    int ret = db->Init();
    if (ret != ErrorInfo::OK) {
        zlog_w("cluster server is offline");
    }

    return ret;
}

void TransportLayer::SyncData() {

    long long start = Util::TimeStamp();
    this->BatchData();
    this->TransferData();
    long long end = Util::TimeStamp();
    zlog_w("Sync data period time %lld ms", end - start);

    return;
}

void TransportLayer::BatchData() {
    XXH64_hash_t hash;
    int          group;
    hashTab.clear();
    hashTab.resize(Configuration::dataConf.begin()->HashGroupCount);

    // 遍历所有readData 进行分组
    for (std::unordered_map<std::string, std::string>::const_iterator iter =
             readData.begin();
         iter != readData.end(); iter++) {
        hash  = XXH64(iter->first.c_str(), iter->first.size(),
                      Configuration::HASHSEED);
        group = hash % Configuration::dataConf.begin()->HashGroupCount;
        hashTab[group].emplace(iter->first, iter->second);
    }
}

void TransportLayer::TransferData() {

    // 遍历所与hash group 发送
    for (int i = 0; i < Configuration::dataConf.begin()->HashGroupCount; ++i) {
        std::string hashKey = Configuration::dataConf.begin()->HashKeyPrefix;
        hashKey.append(std::to_string(i));
        db->UploadHandler(hashKey, hashTab[i]);
        std::this_thread::sleep_for(std::chrono::milliseconds(
            Configuration::dataConf.begin()->DataSyncUpdateInterval));
    }
}

int TransportLayer::OutputData(
    std::unordered_map<std::string, std::string> &controlData) {

    int ret = ErrorInfo::OK;

    ret = db->StreamRecv(controlData);
    if (controlData.empty()) {
        zlog_w("TransportLayer outputdata null");
        return ErrorInfo::EMPTY_DATA;
    }
    return ret;
}

int TransportLayer::OutputData(void (*Callback)(
    std::unordered_map<std::string, std::string> controlData)) {

    int ret = ErrorInfo::OK;

    std::unordered_map<std::string, std::string> writeData;
    ret = db->StreamRecv(writeData);
    if (writeData.empty()) {
        zlog_w("TransportLayer outputdata null");
        return ErrorInfo::EMPTY_DATA;
    }
    Callback(writeData);
    return ret;
}

void TransportLayer::Run() {
    for (;;) {
        this->SyncData();
    }
}
